package com.gitee.loyo;

import com.fasterxml.jackson.annotation.JsonCreator;
import com.fasterxml.jackson.annotation.JsonProperty;
import com.gitee.loyo.impl.DelegateReadStream;
import com.gitee.loyo.utils.HttpUtils;
import com.gitee.loyo.utils.Pair;
import io.vertx.core.AsyncResult;
import io.vertx.core.Future;
import io.vertx.core.Vertx;
import io.vertx.core.buffer.Buffer;
import io.vertx.core.http.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Iterator;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;

public class Endpoints {
    private final static Logger log = LoggerFactory.getLogger(Endpoints.class);
    public final static Map<String, Endpoint> endpoints = new ConcurrentHashMap<>();

    public static boolean registerClient(Vertx vertx, String clientId, ServerWebSocket webSocket) {
        synchronized (endpoints){
            if(endpoints.containsKey(clientId)){
                return false;
            }
            Client bridge = new Client(vertx, clientId, webSocket);
            endpoints.put(clientId, bridge);
            log.info("add new bridge {}", clientId);
            webSocket.frameHandler(bridge::handleResponseFrame).closeHandler(event -> {
                endpoints.remove(clientId);
                bridge.close();
                log.info("remove bridge {}", clientId);
            });
            return true;
        }
    }

    public static Future<Void> addProxy(Proxy proxy){
        synchronized (endpoints){
            if(endpoints.containsKey(proxy.id)){
                return Future.failedFuture(proxy.id + " is exists");
            }
            endpoints.put(proxy.id, proxy);
            for (int i = 0; i < BridgeConfig.INSTANCE.proxies.size(); i++) {
                Endpoints.Proxy record = BridgeConfig.INSTANCE.proxies.get(i);
                if(record.id.equals(proxy)){
                    BridgeConfig.INSTANCE.proxies.set(i, proxy);
                    return BridgeConfig.INSTANCE.save();
                }
            }
            BridgeConfig.INSTANCE.proxies.add(proxy);
            return BridgeConfig.INSTANCE.save();
        }
    }

    public static Future<Void> updateProxy(Proxy proxy){
        synchronized (endpoints){
            if(!endpoints.containsKey(proxy.id)){
                return Future.failedFuture(proxy.id + " is not exists");
            }
            endpoints.put(proxy.id, proxy);
            for (int i = 0; i < BridgeConfig.INSTANCE.proxies.size(); i++) {
                Endpoints.Proxy record = BridgeConfig.INSTANCE.proxies.get(i);
                if(record.id.equals(proxy.id)){
                    BridgeConfig.INSTANCE.proxies.set(i, proxy);
                    return BridgeConfig.INSTANCE.save();
                }
            }
            BridgeConfig.INSTANCE.proxies.add(proxy);
            return BridgeConfig.INSTANCE.save();
        }
    }

    public static Future<Void> removeProxy(Proxy proxy){
        endpoints.remove(proxy.id);
        Iterator<Endpoints.Proxy> ite = BridgeConfig.INSTANCE.proxies.iterator();
        while (ite.hasNext()){
            Endpoints.Proxy record = ite.next();
            if(record.id.equals(proxy.id)){
                ite.remove();
                break;
            }
        }
        return BridgeConfig.INSTANCE.save();
    }

    public static boolean contains(String host) {
        return endpoints.containsKey(host);
    }

    public static void handle(Vertx vertx, Bridges.Bridge bridge, HttpServerRequest request) {
        endpoints.get(bridge.hostname).handleRequest(vertx, request);
    }

    public static class Client extends Endpoint {
        final Vertx vertx;
        final ServerWebSocket webSocket;
        final long timerId;
        final Map<String, Pair<HttpServerRequest, Logs.RequestLog>> unanswered = new ConcurrentHashMap<>();
        long lastHeartbeatTime = System.currentTimeMillis();

        private Client(Vertx vertx, String id, ServerWebSocket webSocket) {
            super(id);
            this.vertx = vertx;
            this.webSocket = webSocket;
            timerId = vertx.setPeriodic(10 * 1000, event -> {
                checkAliveConnection();
                checkAliveClient();
            });
        }

        @Override
        void handleRequest(Vertx vertx, HttpServerRequest request) {
            request.pause();

            Logs.RequestLog requestLog = new Logs.RequestLog();
            requestLog.in.init(request);
            Logs.addLog(request.host(), requestLog);

            String requestId = UUID.randomUUID().toString();
            unanswered.put(requestId, new Pair<>(request, requestLog));


            Future.<Void>future(p ->
                // 往客户端写入request line
                webSocket.write(HttpUtils.wrapReqLine(requestId, request.method(), request.uri())).onComplete(p)
            ).compose(unused ->
                // 往客户端写入header
                webSocket.write(HttpUtils.wrapHeader(requestId, request.headers()))
            ).onComplete(ar -> {
                if(ar.failed()){
                    log.error("write error", ar.cause());
                    unanswered.remove(requestId);
                    request.resume().response().setStatusCode(503).setStatusMessage("Service Unavailable").end();
                    return;
                }

                request.exceptionHandler(t ->
                    log.error("client " + id + " error", log)
                ).endHandler(avoid ->
                    // 通知客户端请求完成
                    webSocket.write(HttpUtils.wrapEnd(requestId)).onComplete(this::handleWriteResult)
                ).handler(data -> {
                    // 往客户端写入body
                    webSocket.write(HttpUtils.wrapChunk(requestId, data)).onComplete(this::handleWriteResult);
                    requestLog.in.body(data);
                });
                request.resume();
            });
        }

        private void handleResponseFrame(WebSocketFrame frame) {
            if(frame.type() == WebSocketFrameType.PING){
                webSocket.writeFrame(WebSocketFrame.pongFrame(Buffer.buffer((byte)0)));
                lastHeartbeatTime = System.currentTimeMillis();
                return;
            }
            if(frame.type() != WebSocketFrameType.BINARY){
                return;
            }
            Buffer pack = frame.binaryData();
            String requestId = HttpUtils.unwrapRequestId(pack);
            Pair<HttpServerRequest, Logs.RequestLog> pair = unanswered.get(requestId);
            if(pair == null){
                return;
            }

            HttpServerRequest request = pair.left;
            Logs.RequestLog requestLog = pair.right;
            HttpServerResponse response = request.response();
            requestLog.out.init(response);
            response.endHandler(event -> requestLog.end());

            HttpUtils.unwrapResponsePack(
                pack,
                response::setStatusCode,
                response::setStatusMessage,
                entry -> response.headers().add(entry.getKey(), entry.getValue()),
                data -> {
                    Buffer buffer = Buffer.buffer(data);
                    response.write(buffer);
                    requestLog.out.body(buffer);
                },
                end -> {
                    response.end();
                    requestLog.end();
                    unanswered.remove(requestId);
                }
            );
        }

        @Override
        public String toString() {
            return "Client{" + id + "}";
        }

        private void checkAliveConnection() {
            log.trace("check alive connection, unanswered size: " + unanswered.size());
            for (Map.Entry<String, Pair<HttpServerRequest, Logs.RequestLog>> entry : unanswered.entrySet()) {
                if(!Global.ALIVE_CONNECTION.contains(entry.getValue().left.connection())){
                    log.warn("close request " + entry.getKey());
                    unanswered.remove(entry.getKey());
                    webSocket.write(HttpUtils.wrapClose(entry.getKey())).onComplete(this::handleWriteResult);
                }
            }
        }

        private void checkAliveClient(){
            if(System.currentTimeMillis() - lastHeartbeatTime > 30 * 1000){
                log.warn("client {} connection timeout", id);
                webSocket.close();
            }
        }

        public void close() {
            vertx.cancelTimer(timerId);
            checkAliveConnection();
            for (Pair<HttpServerRequest, Logs.RequestLog> pair : unanswered.values()) {
                pair.left.response().setStatusCode(444).setStatusMessage("Connection Closed Without Response").end();
                pair.right.out.init(pair.left.response());
                pair.right.end();
            }
        }
    }

    public static class Proxy extends Endpoint {
        public final String host;
        public final int port;
        private HttpClient httpClient;

        @JsonCreator
        private Proxy(@JsonProperty("id") String id, @JsonProperty("host") String host, @JsonProperty("port") int port) {
            super(id);
            this.host = host;
            this.port = port;
        }

        @Override
        void handleRequest(Vertx vertx, HttpServerRequest request) {
            request.pause();
            if(httpClient == null){
                synchronized (this){
                    if(httpClient == null){
                        httpClient = vertx.createHttpClient(new HttpClientOptions().setDefaultHost(host).setDefaultPort(port));
                    }
                }
            }

            Logs.RequestLog requestLog = new Logs.RequestLog();
            requestLog.in.init(request);
            Logs.addLog(request.host(), requestLog);
            DelegateReadStream delegateRequestReadStream = new DelegateReadStream(request, buffer -> requestLog.in.body(buffer));

            RequestOptions proxyRequestOptions = new RequestOptions().setMethod(request.method()).setURI(request.uri()).setHeaders(request.headers());
            httpClient.request(proxyRequestOptions).onComplete(ar -> {
                if(ar.failed()){
                    log.error("proxy error", ar.cause());
                    request.resume().response().setStatusCode(503).setStatusMessage(ar.cause().getMessage()).end();
                    requestLog.out.init(request.response());
                    return;
                }

                HttpClientRequest proxyRequest = ar.result();

                proxyRequest.send(delegateRequestReadStream).onComplete(proxyAr -> {
                    if(proxyAr.failed()){
                        log.error("proxy error", ar.cause());
                        request.response().setStatusCode(503).setStatusMessage(ar.cause().getMessage()).end();
                        requestLog.out.init(request.response());
                        return;
                    }

                    HttpClientResponse proxyResponse = proxyAr.result();
                    DelegateReadStream delegateResponseReadStream = new DelegateReadStream(proxyResponse, buffer -> requestLog.out.body(buffer));
                    requestLog.out.init(proxyResponse);

                    HttpServerResponse response = request.response();
                    response.endHandler(event -> requestLog.end());
                    response.setStatusCode(proxyResponse.statusCode()).setStatusMessage(proxyResponse.statusMessage());
                    for (Map.Entry<String, String> header : proxyResponse.headers()) {
                        response.putHeader(header.getKey(), header.getValue());
                    }
                    response.send(delegateResponseReadStream);
                });
                request.resume();
            });
        }

        public String getHost() {
            return host;
        }

        public int getPort() {
            return port;
        }
    }

    public abstract static class Endpoint {
        public final String id;

        private Endpoint(String id) {
            this.id = id;
        }

        abstract void handleRequest(Vertx vertx, HttpServerRequest request);

        public void handleWriteResult(AsyncResult<Void> ar) {
            if(ar.failed())
                log.error("write error", ar.cause());
        }

        public String getId() {
            return id;
        }
    }
}
